package zhongche;

import com.alibaba.fastjson.JSONObject;
import com.zhongche.config.BackgroundConfig;
import com.zhongche.util.CommUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class KafkaTest {


    public static Properties createKafkaProducerProperties() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.154.100:9092,192.168.154.101:9092,192.168.154.102:9092");
        props.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        return props;
    }

    public static Properties createKafkaConsumerProperties() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.154.100:9092,192.168.154.101:9092,192.168.154.102:9092");
        props.put("key.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        return props;
    }




    public static void sendKafkaData(String topic) {
        Properties pops = createKafkaProducerProperties();
        KafkaProducer<String,String> kafkaProducer = new KafkaProducer<String, String>(pops);
        String jsonStr = "{\n" +
                "  \"box_num\": 1,\n" +
                "  \"current_station\": 5,\n" +
                "  \"dDevice\": 50,\n" +
                "  \"dataParam\": {\n" +
                "\"s0000108\": 0,\n" +
                "\"s0915691\": 0,\n" +
                " \"s1999915\": 1,\n" +
                " \"s0847918\": 6\n" +
                "  },\n" +
                "  \"data_sDeviceNo\": 20,\n" +
                "  \"frame_num\": 5657,\n" +
                "  \"line_id\": 2,\n" +
                "  \"line_num\": \"2\",\n" +
                "  \"message_type\": 5025,\n" +
                "  \"netLink_flag\": 0,\n" +
                "  \"next_station\": 0,\n" +
                "  \"ocs_time\": \"210716153627\",\n" +
                "  \"sDevice\": 40,\n" +
                "  \"sDeviceTime\": \"210716163626\",   \n" +
                "  \"second_circle\": 1,\n" +
                "  \"speed\": 0.0,\n" +
                "  \"start_station\": 5,\n" +
                "  \"terminal_station\": 28,\n" +
                "  \"train_id\": 159160,\n" +
                "  \"train_name\": \"159160\",\n" +
                "  \"train_num\": 159160,\n" +
                "  \"train_type\": \"A52\",\n" +
                "  \"ttype_id\": 10\n" +
                "}\n";
        JSONObject jsonObject = JSONObject.parseObject(jsonStr);
        jsonObject.put("sDeviceTime", "210716163626");
        JSONObject dataParam = jsonObject.getJSONObject("dataParam");
        dataParam.put("s0915691", 70);
        dataParam.put("s0847918", 1);
        String key = CommUtil.md5String(dataParam.toJSONString());
        ProducerRecord<String,String> producerRecord = new ProducerRecord<String,String>(topic, key, jsonObject.toJSONString());
        kafkaProducer.send(producerRecord);
        kafkaProducer.flush();
        kafkaProducer.close();
    }

    public static void kafkaConsumerTest(String groupId, String topic){
        Properties pops = createKafkaConsumerProperties();
        pops.setProperty("group.id",groupId);
        pops.setProperty("enable.auto.commit", "true");
        pops.put("auto.commit.interval.ms", "1000");

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(pops);
        kafkaConsumer.subscribe(Arrays.asList(topic));
        while (true){
            ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(10));
            for (ConsumerRecord<String,String> record: records) {
                System.out.println(record);
            }
            
        }
        }

    




    public static void main(String[] args) {
//        sendKafkaData(BackgroundConfig.KAFKA_TOPIC_GUANGZHOU_2);
        kafkaConsumerTest("test","test_topic");
    }

}
